package com.exceeddata.vcloud.common.kafka.consumer;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.exceeddata.vcloud.business.domain.dto.PushKafkaDto;
import lombok.extern.log4j.Log4j;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

/**
 * @author: yxh
 * @Description: Kafka消费者
 * @Date: Created in 2025/3/3 14:54
 * @Modified By:
 */
@Service
@Slf4j
public class KafkaConsumer {

    @KafkaListener(topics = "vhr-file-info-snapshot-01",containerFactory = "batchFactory")
    public void listen(String message, Acknowledgment acknowledgment) {
        System.err.println("消费者{}"+message);
        PushKafkaDto dto = JSONObject.parseObject(message, PushKafkaDto.class);
        System.err.println("消费者对象{}"+JSON.toJSONString(dto));
        //如果你的方法参数是一个 String 类型，那么默认情况下，这个参数会被解析为消息的值（Value）
        log.info("Received Message: {}" , message);
        // 手动提交偏移量
        acknowledgment.acknowledge();
    }


    /**
     * 获取消息的键（Key）和值（Value），可以使用 ConsumerRecord 对象。ConsumerRecord 包含了消息的所有信息，包括键、值、分区、偏移量等。
     */
    @KafkaListener(topics = "vhr-file-info-snapshot-01",containerFactory = "batchFactory")
    public void listen(ConsumerRecord<String, String> record) {
        System.err.println(record.key());
        log.info("Received Message Key: " + record.key());
        log.info("Received Message Value: " + record.value());
        log.info("Partition: " + record.partition());
        log.info("Offset: " + record.offset());
    }
}
